package com.boarsoft.boar.batch.service.impl;

import com.boarsoft.boar.batch.biz.BatchEventBiz;
import com.boarsoft.boar.batch.biz.BatchExecutionBiz;
import com.boarsoft.boar.batch.biz.BatchInfoBiz;
import com.boarsoft.boar.batch.entity.BatchEvent;
import com.boarsoft.boar.batch.entity.BatchExecution;
import com.boarsoft.boar.batch.entity.BatchInfo;
import com.boarsoft.boar.batch.exception.DeregisterBatchException;
import com.boarsoft.boar.batch.service.*;
import com.boarsoft.common.Util;
import com.boarsoft.common.util.InetUtil;
import com.boarsoft.common.util.RandomUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Date;

/**
 *
 * 简化版本的 batchDispatcher 不登记 batch_execution和batch_event
 *
 * 参考 BatchDispatcherImpl
 */
public class SimpleBatchDispatcherImpl implements BatchCallback, BatchDispatcher {

	private final static Logger log = LoggerFactory.getLogger(SimpleBatchDispatcherImpl.class);

	@Autowired
	protected BatchExecutionBiz executionBiz;
	@Autowired
	protected BatchInfoBiz infoBiz;
	@Autowired
	protected BatchScheduler batchScheduler;
	@Autowired
	protected BatchInfoService batchInfoSvc;
	/** 负责执行批量的远程接口 */
	// @Autowired
	protected BatchBrokerService broker;
	/** */
	@Autowired
	protected BatchEventBiz eventBiz;

	@Override
	public int start(String infoId) {
		return this.start(infoId, null, null);
	}

	@Override
	public int start(String infoId, String params) {
		return this.start(infoId, null, params);
	}

	@Override
	public int start(String infoId, String entry, String params) {
		if (Util.strIsEmpty(infoId)) {
			log.error("Batch info id can not be empty");
			return 1;
		}
		BatchInfo info = infoBiz.get(infoId);
		if (info == null) {
			log.error("Batch info {} not found", infoId);
			return 2;
		}
		//
		// String key = info.toString();
		Date startTime = new Date();

		// 前次执行必须完成且成功

		// 没有传入参数就使用默认参数
		if (Util.strIsEmpty(params)) {
			params = info.getParams();
		}
		// 登记批量执行记录
		BatchExecution exe = new BatchExecution();
		exe.setAddr(InetUtil.getAddr());
		exe.setCatalog(info.getCatalog());
		exe.setCode(info.getCode());
		exe.setEntry(entry);
		exe.setInfoId(infoId);
		// exe.setNo(no);
		exe.setParams(params);
		exe.setStartTime(startTime);
		exe.setStatus(BatchExecution.STATUS_NEW);
		exe.setType(BatchExecution.TYPE_START);
		exe.setBatchId(RandomUtil.randomUUID());
//		String exeId = executionBiz.save(exe);

		// 开始执行
		log.info("Start batch execution {}", exe);

		// 记录事件

		//
		try {
			String addr = (String) broker.start(exe);
//			log.info("Batch execution {} be post to {}", exe, addr);
//			exe.setStatus(BatchExecution.STATUS_QUEUED);
//			exe.setAddr(addr);
//			executionBiz.update(exe);
			return 0;
		} catch (Throwable e) {
//			log.error("Error on post batch execution {}", exeId, e);
//			exe.setStatus(BatchExecution.STATUS_FAILED);
//			exe.setError(e.getMessage());
//			executionBiz.update(exe);
			return 6;
		}
	}

	@Override
	public int resume(String exeId) {
		// 获取要续跑的这条记录
		BatchExecution exe = executionBiz.get(exeId);
		if (exe == null) {
			return BatchExecution.ERROR_NOT_FOUND;
		}
		// 只有失败、挂起或中断的任务组执行才可以续跑
		short status = exe.getStatus();
		if (BatchExecution.STATUS_FAILED != status //
				&& BatchExecution.STATUS_SUSPEND != status//
				&& BatchExecution.STATUS_STOPPED != status) {
			return BatchExecution.ERROR_WRONG_STATUS;
		}
		BatchInfo info = infoBiz.get(exe.getInfoId());
		if (info == null) {
			return BatchExecution.ERROR_MISSING_INFO;
		}

		// 更新批量组执行记录
		// exe.setParams(params); // 保持原有参数
		exe.setAddr(InetUtil.getAddr());
		exe.setStartTime(new Date());
		exe.setStatus(BatchExecution.STATUS_RUNNING);
		exe.setType(BatchExecution.TYPE_RESUME);
		executionBiz.save(exe);

		// 开始执行
		log.info("Resume batch execution {}", exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("resume");
		evt.setExeId(exe.getId());
		// evt.setParams(params);
		evt.setStatus(BatchExecution.STATUS_RUNNING);
		// evt.setStep(step);
		eventBiz.save(evt);

		try {
			String addr = broker.resume(exe);
			exe.setAddr(addr);
			executionBiz.save(exe);
			// return new ReplyInfo<String>();
			return 0;
		} catch (Throwable e) {
			log.error("Error on resume batch execution {}", exeId, e);
			exe.setStatus(BatchExecution.STATUS_FAILED);
			exe.setError(e.getMessage());
			exe.setEndTime(new Date());
			executionBiz.save(exe);
			return BatchExecution.ERROR_UNKNOWN;
		}
	}

	@Override
	public int redo(String exeId) {
		// 获取要重跑的这条记录
		BatchExecution exe = executionBiz.get(exeId);
		return this.redo(exe);
	}

	public int redo(BatchExecution exe) {
		log.info("Redo batch exection {}", exe);
		if (exe == null) {
			return BatchExecution.ERROR_NOT_FOUND;
		}
		// 只有成功、失败、挂起或中断的批量任务才可以重跑
		short status = exe.getStatus();
		if (BatchExecution.STATUS_SUCCESS != status//
				&& BatchExecution.STATUS_FAILED != status//
				&& BatchExecution.STATUS_SUSPEND != status//
				&& BatchExecution.STATUS_STOPPED != status) {
			return BatchExecution.ERROR_WRONG_STATUS;
		}

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("redo");
		evt.setExeId(exe.getId());
		// evt.setParams(params);
		evt.setStatus(BatchExecution.STATUS_RUNNING);
		// evt.setStep(step);
		eventBiz.save(evt);

		// 更新批量组执行记录
		// exe.setParams(bp.getParams()); 保持原有参数
		exe.setAddr(InetUtil.getAddr());
		exe.setStartTime(new Date());
		exe.setStatus(BatchExecution.STATUS_RUNNING);
		exe.setType(BatchExecution.TYPE_REDO);
		executionBiz.save(exe);

		try {
			String addr = broker.redo(exe);
			exe.setAddr(addr);
			executionBiz.save(exe);
			// return new ReplyInfo<String>();
			return 0;
		} catch (Exception e) {
			log.error("Error on redo batch execution {}", exe, e);
			exe.setStatus(BatchExecution.STATUS_FAILED);
			exe.setError(e.getMessage());
			exe.setEndTime(new Date());
			executionBiz.save(exe);
			return BatchExecution.ERROR_UNKNOWN;
		}
	}

	@Override
	public int cancel(String exeId, boolean force) {
		log.info("Try cancel batch execution {}.", exeId);
		// 取消某批量的执行，如果该批量状态为执行中、挂起中则改为取消中。
		BatchExecution exe = executionBiz.get(exeId);
		if (exe == null) {
			return BatchExecution.ERROR_NOT_FOUND;
		}
		// 只有初始0、排队1或执行2的任务组执行才可以取消
		short status = exe.getStatus();
		if (BatchExecution.STATUS_NEW != status//
				&& BatchExecution.STATUS_QUEUED != status//
				&& BatchExecution.STATUS_RUNNING != status) {
			return BatchExecution.ERROR_WRONG_STATUS;
		}

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("cancel");
		evt.setExeId(exeId);
		evt.setStatus(exe.getStatus());
		evt.setStep(exe.getPhrase());
		eventBiz.save(evt);

		// 将执行中、续跑中的批量执行记录改为取消中
		if (status == BatchExecution.STATUS_RUNNING) {
			exe.setStatus(BatchExecution.STATUS_STOPPING);
			executionBiz.save(exe);
			// 通知批量的执行者，停止执行
			try {
				broker.cancel(exe, force);
			} catch (Exception e) {
				log.error("Error on cancel batch execution {}", exe);
				return BatchExecution.ERROR_UNKNOWN;
			}
		} else {
			exe.setStatus(BatchExecution.STATUS_STOPPED);
			executionBiz.save(exe);
		}
		return 0;
	}

	@Override
	public void onStep(String exeId, String step, String addr) {
		log.info("Batch execution {} step {} at {}", exeId, step, addr);
		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchExecution.STATUS_RUNNING);
		exe.setPhrase(step);
		exe.setAddr(addr);
		executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("step");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		evt.setStep(step);
		eventBiz.save(evt);
	}

	@Override
	public void onSuccess(String exeId, String addr) {
		log.info("Batch execution {} completed at {}", exeId, addr);
		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchExecution.STATUS_SUCCESS);
		exe.setPhrase(null);
		exe.setAddr(addr);
		exe.setEndTime(new Date());
		executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("success");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		// evt.setStep(step);
		eventBiz.save(evt);
	}

	@Override
	public void onSuspend(String exeId, String step, String addr) {
		log.info("Batch execution {} suspend {} at {}", exeId, step, addr);

		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchExecution.STATUS_SUSPEND);
		exe.setPhrase(step);
		exe.setAddr(addr);
		executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("suspend");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		// evt.setStep(step);
		eventBiz.save(evt);
	}

	@Override
	public void onFailed(String exeId, String step, String addr, String error, String params) {
		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		if (DeregisterBatchException.MESSAGE.equals(error)) {
			log.warn("Deregister batch {}", exe.getInfoId());
			batchInfoSvc.deregister(exe.getInfoId());
			return;
		}

		log.info("Batch execution {}/{} failed at {}", exeId, step, addr);
		exe.setPhrase(step);
		exe.setAddr(addr);
		exe.setStatus(BatchExecution.STATUS_FAILED);
		exe.setEndTime(new Date());
		exe.setError(error);
		executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("failed");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStep(step);
		evt.setStatus(exe.getStatus());
		eventBiz.save(evt);

		// TODO Failable flag
		BatchInfo info = infoBiz.get(exe.getInfoId());
		if (info == null) {
			log.error("Batch info of {} not exists", exe);
			return;
		}
		int ff = info.getFailable() % 10;
		switch (ff) {
		case BatchInfo.FAILABLE_STOP: // 取消可能还在执行的STEP
			try {
				broker.cancel(exe, true);
			} catch (Exception e) {
				log.error("Failed to cancel {}", exe);
			}
			break;
		case BatchInfo.FAILABLE_IGNORE: // 仅登记事件，不改状态，不做处理
			break;
		case BatchInfo.FAILABLE_RESUME:
			this.resume(exeId);
		case BatchInfo.FAILABLE_REDO:
			this.redo(exe);
		}
	}

	@Override
	public void onStop(String exeId, String step, String addr) {
		// 被stop的批量仅记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(addr);
		evt.setContent("stop");
		evt.setExeId(exeId);
		evt.setStatus(BatchExecution.STATUS_STOPPED);
		evt.setStep(step);
		eventBiz.save(evt);
	}

	@Override
	public void onTimeout(String exeId, String step, String addr) {
		// 此方法是客户端明确通知执行超时用的
		// 记录批量超时事件（由批量应用端主动通知）
		BatchEvent evt = new BatchEvent();
		evt.setAddr(addr);
		evt.setContent("timeout");
		evt.setExeId(exeId);
		evt.setStatus(BatchExecution.STATUS_TIMEOUT);
		evt.setStep(step);
		eventBiz.save(evt);
	}

	@Override
	public void onBegin(String exeId, String addr) {
		// 记录批量超时事件（由批量应用端主动通知）
		BatchEvent evt = new BatchEvent();
		evt.setAddr(addr);
		evt.setContent("begin");
		evt.setExeId(exeId);
		evt.setStatus(BatchExecution.STATUS_RUNNING);
		eventBiz.save(evt);
	}

	public BatchExecutionBiz getExecutionBiz() {
		return executionBiz;
	}

	public void setExecutionBiz(BatchExecutionBiz executionBiz) {
		this.executionBiz = executionBiz;
	}

	public BatchInfoBiz getInfoBiz() {
		return infoBiz;
	}

	public void setInfoBiz(BatchInfoBiz infoBiz) {
		this.infoBiz = infoBiz;
	}

	public BatchBrokerService getBroker() {
		return broker;
	}

	public void setBroker(BatchBrokerService broker) {
		this.broker = broker;
	}

	public BatchEventBiz getEventBiz() {
		return eventBiz;
	}

	public void setEventBiz(BatchEventBiz eventBiz) {
		this.eventBiz = eventBiz;
	}

	public BatchScheduler getBatchScheduler() {
		return batchScheduler;
	}

	public void setBatchScheduler(BatchScheduler batchScheduler) {
		this.batchScheduler = batchScheduler;
	}
}
